package video

import (
	"app-api/app/internal/consumers"
	"app-api/common/constant"
	"context"
	"encoding/json"
	"fmt"
	"github.com/apache/pulsar-client-go/pulsar"

	"app-api/app/internal/svc"
	"app-api/app/internal/types"

	"github.com/zeromicro/go-zero/core/logx"
)

type PostFavVideoLogic struct {
	logx.Logger
	ctx    context.Context
	svcCtx *svc.ServiceContext
}

func NewPostFavVideoLogic(ctx context.Context, svcCtx *svc.ServiceContext) *PostFavVideoLogic {
	return &PostFavVideoLogic{
		Logger: logx.WithContext(ctx),
		ctx:    ctx,
		svcCtx: svcCtx,
	}
}

func (l *PostFavVideoLogic) PostFavVideo(req *types.VideoIDRequest) (resp *types.Empty, err error) {
	userID, _ := l.ctx.Value("id").(json.Number).Int64()
	topicKey := constant.UserLikeVideoTopic
	topicVal := fmt.Sprintf("%d:%d", userID, req.VideoID)
	producer, err := l.svcCtx.PulsarClient.CreateProducer(pulsar.ProducerOptions{Topic: topicKey})
	if err != nil {
		l.Logger.Error("Failed to create producer: %v", err)
		return nil, err
	}

	messageID, err := producer.Send(l.ctx, &pulsar.ProducerMessage{Payload: []byte(topicVal)})
	if err != nil {
		l.Logger.Error("Failed to send message: %v", err)
		return nil, err
	}
	l.Logger.Info("Sent message %s with message ID: %s\n", topicVal, messageID)

	logic := consumers.NewDConsumerLogic(context.Background(), l.svcCtx)
	go consumers.UserVideoConsumer(logic, topicKey)
	return &types.Empty{}, nil
}
